package raft

import "sync/atomic"

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//

func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).

	rf.mu.Lock()
	defer rf.mu.Unlock()

	state := atomic.LoadInt32(&rf.state)
	isLeader = state == Leader
	term = rf.currentTerm
	if isLeader {
		index = rf.getLastLogIndex() + 1
		entry := Entry{
			Index:   index,
			Term:    rf.currentTerm,
			Command: command,
		}
		rf.log = append(rf.log, entry)
		rf.persist()
		DPrintf("[%d] leader append %d entry %+v", rf.me, index, entry)
	}

	return index, term, isLeader
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

func (rf *Raft) attempAppendEntries() {
	rf.mu.Lock()
	term := rf.currentTerm
	leaderId := rf.me
	leaderCommit := rf.commitIndex
	rf.mu.Unlock()

	for i := range rf.peers {
		if i == rf.me {
			continue
		}

		go func(server int) {

			rf.mu.Lock()
			entries := []Entry{}
			nextIndex := rf.nextIndex[server]
			if nextIndex < len(rf.log) {
				// end := nextIndex + 10
				end := len(rf.log)
				if end > len(rf.log) {
					end = len(rf.log)
				}
				entries = rf.log[nextIndex:end]
			}

			prevLogIndex := nextIndex - 1
			prevLogTerm := rf.log[prevLogIndex].Term
			rf.mu.Unlock()

			args := &AppendEntriesArgs{
				Term:         term,
				LeaderId:     leaderId,
				PrevLogIndex: prevLogIndex,
				PrevLogTerm:  prevLogTerm,
				Entries:      entries,
				LeaderCommit: leaderCommit,
			}

			reply := &AppendEntriesReply{}

			// DPrintf("[%d] leader at term %d send %d follower %v entries, the nextIndex is %d", rf.me, term, server, len(entries), nextIndex)
			ok := rf.sendAppendEntries(server, args, reply)
			if !ok {
				// DPrintf("[%d] send heartbeat to %d at term %d fail", rf.me, server, term)
				return
			}

			rf.mu.Lock()
			defer rf.mu.Unlock()
			if reply.Term > rf.currentTerm {
				rf.currentTerm = reply.Term
				state := atomic.LoadInt32(&rf.state)
				atomic.StoreInt32(&rf.state, Follower)
				rf.persist()
				if state == Leader {
					rf.tickerCh <- struct{}{}
				}
				DPrintf("[%d] leader convert to follower", rf.me)
			} else if atomic.LoadInt32(&rf.state) == Leader && term == rf.currentTerm {
				rf.nextIndex[server] = Min(reply.LastLogIndex+1, len(rf.log))
				DPrintf("[%d] leader args %+v nextIndex is %+v", rf.me, args, rf.nextIndex)
				if reply.Success {
					rf.matchIndex[server] = reply.LastLogIndex

					appendIndex := reply.LastLogIndex

					if rf.commitIndex < appendIndex && appendIndex < len(rf.log) && rf.log[appendIndex].Term == rf.currentTerm {
						appended := 1

						for i := 0; i < len(rf.matchIndex); i++ {
							if i == rf.me {
								continue
							}

							if rf.matchIndex[i] >= appendIndex {
								appended++
							}
						}

						if appended > len(rf.peers)/2 {
							rf.commitIndex = appendIndex
						}

					}

					if rf.commitIndex > rf.lastApplied {
						for i := rf.lastApplied + 1; i <= rf.commitIndex; i++ {
							rf.applyCh <- ApplyMsg{
								CommandValid: true,
								Command:      rf.log[i].Command,
								CommandIndex: i,
							}
						}
						rf.lastApplied = rf.commitIndex
						DPrintf("[%d] leader applied %d entry %+v", rf.me, rf.lastApplied, rf.log[rf.lastApplied])
					}
				}
			}
		}(i)
	}
}
